package hi.storm.test.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.topology.BoltDeclarer;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import hi.storm.test.blot.ShowWordBolt;
import hi.storm.test.blot.SplitBolt;
import hi.storm.test.blot.SumBolt;
import hi.storm.test.spout.TestSpout;

/**
 *
 * Title: App Description: storm测试 Version:1.0.0
 *
 * @author pancm
 * @date 2018年3月6日
 */
public class App {
	private static final String test_spout = "test_spout";
	private static final String word_bolt = "word_bolt";
	private static final String test_bolt = "split_bolt";
	private static final String test2_bolt = "sum_bolt";

	public static void main(String[] args) {
		// 定义一个拓扑
		StormTopology topology = createTopology();
		Config conf = new Config();
		conf.put("test", "test");
		try {
			// 运行拓扑
			if (args != null && args.length > 0) { // 有参数时，表示向集群提交作业，并把第一个参数当做topology名称
				System.out.println("run remote model");
				StormSubmitter.submitTopology(args[0], conf, topology);
			} else {// 没有参数时，本地提交
				// 启动本地模式
				System.out.println("run local model");
				LocalCluster cluster = new LocalCluster();
				cluster.submitTopology("Word-counts", conf, topology);
				Thread.sleep(2000000);
				// //关闭本地集群
				cluster.shutdown();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static StormTopology createTopology() {

		// 定义一个拓扑
		TopologyBuilder builder = new TopologyBuilder();

		TestSpout spout = new TestSpout();
		// 设置一个Executeor(线程)，默认一个
		builder.setSpout(test_spout, spout, 1);
		// shuffleGrouping:表示是随机分组
		// 设置一个Executeor(线程)，和一个task
		IRichBolt blot = new SplitBolt();
		BoltDeclarer declare = builder.setBolt(test_bolt, blot, 1);
		declare = declare.setNumTasks(2);
		declare.shuffleGrouping(test_spout);

		blot = new ShowWordBolt();
		declare =  builder.setBolt(word_bolt, new SumBolt(), 1);
		declare =  declare.setNumTasks(1);
		declare.shuffleGrouping(test_spout);

		// fieldsGrouping:表示是按字段分组
		// 设置一个Executeor(线程)，和一个task
		blot = new SumBolt();
		declare =  builder.setBolt(test2_bolt, new SumBolt(), 1);
		declare =  declare.setNumTasks(1);
		declare =  declare.fieldsGrouping(test_bolt, new Fields("count"));
		return builder.createTopology();
	}

}
